package com.myz.rabbitmq.autoAck;

import com.myz.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 消息消费者1
 *
 * @author maoyz on 18-4-4.
 */
public class Work1 {
    private static final Logger logger = LoggerFactory.getLogger(Work1.class);

    private static final String TASK_QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
        System.out.println("Worker1  Waiting for messages");

        // 每次从队列获取的数量
        // 保证一次只分发一个
        // autoAck是否自动回复，如果为true的话，每次生产者只要发送信息就会从内存中删除，
        // 那么如果消费者程序异常退出，那么就无法获取数据，我们当然是不希望出现这样的情况，
        // 所以才去手动回复，每当消费者收到并处理信息然后在通知生成者。最后从队列中删除这条信息。
        // 如果消费者异常退出，如果还有其他消费者，那么就会把队列中的消息发送给其他消费者，
        // 如果没有，等消费者启动时候再次发送。
        channel.basicQos(1);

        // 定义消费者
        final Consumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");

                logger.debug("Worker1  Received '" + message + "'");

                try {
                    // throw new Exception();
                    doWork(message);
                } catch (Exception e) {
                    channel.abort();
                } finally {
                    logger.debug("Worker1 Done");
                    // 情况一：对处理好的消息进行应答
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        // 关闭自动分发
        boolean autoAck = false;
        // 消息消费完成确认
        channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);
    }

    private static void doWork(String task) {
        try {
            Thread.sleep(1000); // 暂停1秒钟
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
